package org.suyafei.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.suyafei.func.UniqueVisitorFilterFunction;
import org.suyafei.util.KafkaUtil;

/**
 * @author 宿某人
 * @Date: 2022/04/01/16:00
 * @Description:
 */
public class DwmUniqueVisitorApp {

    /**
     * 定义topic
     */
    public static final String SOURCE_TOPIC = "dwm_link_visit_topic";

    /**
     * 消费者组
     */
    public static final String GROUP_ID = "dwm_unique_visitor_group";


    public static final String SINK_TOPIC = "dwm_unique_visitor_topic";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取数据
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);

        DataStreamSource<String> ds = env.addSource(kafkaConsumer);
        //数据转换
        SingleOutputStreamOperator<JSONObject> jsonDS = ds.map(jsonStr -> JSON.parseObject(jsonStr));

        //分组
        KeyedStream<JSONObject, Object> keyedStream = jsonDS.keyBy(new KeySelector<JSONObject, Object>() {
            @Override
            public Object getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("udid");
            }
        });
        //排重过滤
        SingleOutputStreamOperator<JSONObject> filterDS = keyedStream.filter(new UniqueVisitorFilterFunction());
        filterDS.print("独立访客");
        //转成字符串写入kafka

        SingleOutputStreamOperator<String> uniqueVisitorDS = filterDS.map(obj -> obj.toJSONString());

        FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);
        uniqueVisitorDS.addSink(kafkaProducer);
        env.execute();

    }
}
